package com.tt.vtg.scenario.kafka;

import com.tt.vtg.scenario.util.ValueUtils;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.Collections;
import java.util.Properties;
import java.util.function.Consumer;

public class KafkaConsumerTask implements Runnable {
    private static final String TOPIC = "inbound";
    private KafkaConsumer<String, String> consumer;
    private volatile boolean running = true;
    private Consumer<ConsumerRecord<String, String>> action;

    public KafkaConsumerTask(Consumer<ConsumerRecord<String, String>> action){
        this.action = action;

        final Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                ValueUtils.get("kafka.bootstrap.server"));
        props.put(ConsumerConfig.GROUP_ID_CONFIG,
                "order.inbound");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class.getName());
        /*
        *   earliest   当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费。
            latest     当分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费该分区下新产生的的数据。
            none       当该topic下所有分区中存在未提交的offset时，抛出异常。
        * */
        props.put("auto.offset.reset", "latest");

        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList(TOPIC));
    }

    public void stop(){
        running = false;
    }

    @Override
    public void run() {
        while(running) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            records.forEach(record -> {
                action.accept(record);

                System.out.printf("Consumer record:\nkey = %s\nvalue = %s\npartition = %d\noffset = %d\nExec by = %s\n",
                        record.key(), record.value(),
                        record.partition(), record.offset(),
                        Thread.currentThread().getName());
                System.out.println("======================================");
            });
            consumer.commitAsync();
        }
    }
}
